package com.changdu.seatunnel.admin.util;

import com.changdu.seatunnel.admin.entity.DataInstance;
import com.changdu.seatunnel.admin.pojo.SourceConfig;
import com.changdu.seatunnel.admin.pojo.SyncConfigGenerate;
import com.changdu.seatunnel.admin.pojo.TableMapping;
import com.fasterxml.jackson.databind.JsonNode;

public class TiDBToStarrocksTempltes extends BaseTemplate {

	/**
	 * {
	    "env": {
	        "parallelism": 1,
	        "job.mode": "STREAMING",
	        "job.name": "tidb=>starrocks : {{dc_code}}.{{source_database}}.{{source_table}}",
	        "checkpoint.interval": "600000",
	        "checkpoint.timeout": "600000"
	    },
	    "source" : [
	        {
	            "plugin_name": "TiDB-CDC",
	            "pd-addresses": "10.20.10.61:2379",
	            "driver": "com.mysql.cj.jdbc.Driver",
	            "base-url": "jdbc:mysql://{{source_host}}:{{source_port}}/{{source_database}}?useSSL=false",
	            "database-name" : "{{source_database}}",
	            "username": "{{source_username}}",
	            "password": "{{source_password}}",
	            "table-name": "{{source_table}}",
	            "tidb.queue.type": "memory",
	            "tidb.memory.queue.size": "10240",
	            "tikv.grpc.timeout_in_ms": "260000"
	        }
	    ],
	    "transform": [
	        {
	            "query": "select {{transform_sql}} from source_table",
	            "plugin_name": "Sql"
	        }
	    ],
	    "sink": [
	        {
	            "plugin_name": "StarRocks",
	            "schema_dynamic": "true",
	            "data_save_mode": "APPEND_DATA",
	            "batch_max_rows": 40960,
	            "batch_max_bytes": "100663296",
	            "max_retries": "17280",
	            "max_retry_backoff_ms": 5000,
	            "retry_backoff_multiplier_ms": 10000,
	            "nodeUrls": [
	                "{{target_host}}:18030"
	            ],
	            "base-url": "jdbc:mysql://{{target_host}}:19030?useSSL=false",
	            "username": "{{target_username}}",
	            "password": "{{target_password}}",
	            "enable_upsert_delete": "true"
	        }
	    ]
	}
	 * @param json
	 * @param config
	 * @return
	 */
	public static JsonNode createTemplate(DataInstance sourceInstance, DataInstance targetInstance, TableMapping tableMapping, String template, SyncConfigGenerate config) {
		SourceConfig sourceConfig = config.getSourceConfig();
		String[] sourceTable = tableMapping.getSourceTable().split("\\.", 2);
		String[] targetTable = tableMapping.getTargetTable().split("\\.", 2);
		
        String content = template
    		.replace("{{source_dc_code}}", sourceConfig.getDataCenter())
            .replace("{{source_database}}", sourceTable[0])
            .replace("{{source_table}}", sourceTable[1])
            .replace("{{source_regist_host}}", toString(sourceInstance.getRegistHost(), ""))
            .replace("{{source_host}}", toString(sourceInstance.getHost(), ""))
            .replace("{{source_port}}", toString(sourceInstance.getPort(), ""))
            .replace("{{source_database}}", sourceTable[0])
            .replace("{{source_database}}", sourceTable[0])
            .replace("{{source_username}}", toString(sourceInstance.getUsername(), ""))
            .replace("{{source_password}}", toString(sourceInstance.getPassword(), ""))
            .replace("{{transform_sql}}", KafkaToStarrocksTempltes.buildTransform(tableMapping))
            .replace("{{target_regist_host}}", toString(targetInstance.getRegistHost(), ""))
            .replace("{{target_host}}", toString(targetInstance.getHost(), ""))
            .replace("{{target_port}}", toString(targetInstance.getPort(), ""))
            .replace("{{target_username}}", toString(targetInstance.getUsername(), ""))
            .replace("{{target_password}}", toString(targetInstance.getPassword(), ""))
            .replace("{{target_database}}", targetTable[0])
            .replace("{{target_table}}", targetTable[1])
            ;
        return transformIgnore(tableMapping, JsonUtils.getJsonNode(content.getBytes()));
	}

}
